package com.atguigu.cn.dataStream.sink

import java.util.Properties

import com.atguigu.cn.dataStream.api.SensorReading
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}

/**
  * @author: yangShen
  * @Description:
  * @Date: 2020/4/13 16:53 
  */
object KafkaSinkTest {
  def main(args: Array[String]): Unit = {

    //1.environment
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    // 2.source
    //val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")

    //2. 从kafka读取数据
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "10.16.26.16:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    //2.消费者去kafka中读取数据----分区sensor
    val inputStream = environment.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))

    // transform 操作，并标记返回类型
    val dataStream: DataStream[String] = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble).toString //转成string方便序列化输出
    })

    // sink --->  生产者把消费者读到的数据写到（输出:sink）----另一个分区sinkTest
    dataStream.addSink(new FlinkKafkaProducer011[String]("10.16.26.16:9092","sinkTest", new SimpleStringSchema()))
    dataStream.print()

    environment.execute("kafka sink test")

  }

}
